package com.jcooky.tutorfs.proxy.filesystem;

import java.io.EOFException;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;


public class HDFSFileSystem extends AbstractFileSystem {
	private FileSystem fs;

	public void mkdir(String strpath) {
		Path path = new Path(strpath);
	    
		try {
			fs.mkdirs(path);
		} catch(IOException e) {
			throw new FileSystemException(e);
		}
	}
	
	public void rmdir(String strpath) {
		try {
			fs.delete(new Path(strpath), true);
		} catch(IOException e) {
			throw new FileSystemException(e);
		}
	}
	
	public byte[] read(String strpath, int pos, int size) {
		Path path = new Path(strpath);
		FSDataInputStream is = null;
		try {
			byte []bytes = new byte[256];
			if (!fs.exists(path))
				throw new FileSystemException();
			
			is = fs.open(path);
			is.skipBytes(pos);
			try {
				is.read(bytes, 0, size);
			} catch(EOFException e) {
			}
			return bytes;
		} catch(IOException e) {
			throw new FileSystemException(e);
		} finally {
			if (is != null) {
				try { is.close(); } catch(IOException e) {}
			}
		}
	}
	
	public void write(String strpath, byte[] data, int pos, int size) {
		Path path = new Path(strpath);
		FSDataOutputStream os = null;
		try {
			if (fs.exists(path)) {
				os = fs.append(path);
				os.write(data, 0, size);
			} else {
				os = fs.create(path);
				os.write(data, 0, size);
			}
		} catch(IOException e) {
			throw new FileSystemException(e);
		} finally {
			if (os != null) {
				try { os.close(); } catch(IOException e) {}
			}
		}
	}
	
	public void open() {
		Configuration conf = new Configuration(true);
		conf.set("fs.default.name", "hdfs://10.211.55.8:9000");
		conf.set("dfs.replication", "1");
		
		try {
			this.fs = FileSystem.get(conf);
		} catch(IOException e) {
			throw new FileSystemException(e);
		}
	}
	
	public void close() throws FileSystemException {
		try {
			this.fs.close();
		} catch(IOException e) {
			throw new FileSystemException(e);
		}
	}
}
